
package com.jf.cloud.search.feign;


import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.jf.cloud.api.dto.EsPageDTO;
import com.jf.cloud.api.feign.SearchOrderTaskFeignClient;
import com.jf.cloud.api.order.constant.OrderStatus;
import com.jf.cloud.api.order.feign.OrderFeignClient;
import com.jf.cloud.api.order.feign.PurchaseOrderFeignClient;
import com.jf.cloud.api.vo.EsPageVO;
import com.jf.cloud.common.constant.StatusEnum;
import com.jf.cloud.common.exception.LuckException;
import com.jf.cloud.common.order.bo.EsOrderBO;
import com.jf.cloud.common.order.bo.PayNotifyBO;
import com.jf.cloud.common.order.dto.OrderSearchDTO;
import com.jf.cloud.common.order.dto.PurchaseOrderSearchDTO;
import com.jf.cloud.common.order.vo.EsPurchaseOrderVO;
import com.jf.cloud.common.product.constant.PreSaleType;
import com.jf.cloud.common.response.ResponseEnum;
import com.jf.cloud.common.response.ServerResponseEntity;
import com.jf.cloud.common.rocketmq.config.RocketMqConstant;
import com.jf.cloud.search.constant.EsConstant;
import com.jf.cloud.search.manager.OrderSearchManager;
import com.jf.cloud.search.manager.PurchaseOrderSearchManager;
import com.jf.cloud.search.util.EsSearchUtil;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.web.bind.annotation.RestController;
import io.swagger.v3.oas.annotations.Hidden;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * 搜索符合定时任务条件的订单
 * @author zz
 * @date 2020/12/07
 */
@RestController
@Hidden
public class SearchOrderTaskFeignController implements SearchOrderTaskFeignClient {
    private static final Logger logger = LoggerFactory.getLogger(SearchOrderTaskFeignController.class);

    private static final String SETTLED = "settled";
    private static final String CONFIRM = "confirm";
    private static final String PRESALE = "presale";
    private static final String PRESALE_CANAL = "presaleCanal";

    @Autowired
    private OrderSearchManager orderSearchManager;
    @Autowired
    private OrderFeignClient orderFeignClient;
    @Autowired
    private RocketMQTemplate orderNotifyTemplate;
    @Autowired
    private RocketMQTemplate presaleOrderCanalMqTemplate;
    @Autowired
    private PurchaseOrderSearchManager purchaseOrderSearchManager;
    @Autowired
    private PurchaseOrderFeignClient purchaseOrderFeignClient;

    @Override
    public ServerResponseEntity settledOrder() {
        //设置查询超时时间
        OrderSearchDTO param = new OrderSearchDTO();
        param.setStatus(OrderStatus.SUCCESS.value());
        param.setIsSettled(0);
        Date finallyTime = DateUtil.beginOfDay(DateUtil.offsetDay(new Date(), -15)).toJdkDate();
        param.setFinallyTime(finallyTime);
        return handleOrder(param, SETTLED);
    }

    @Override
    public ServerResponseEntity confirmOrder() {
        OrderSearchDTO param = new OrderSearchDTO();
        param.setStatus(OrderStatus.CONSIGNMENT.value());
        Date deliveryTime = DateUtil.beginOfDay(DateUtil.offsetDay(new Date(), -15)).toJdkDate();
        param.setDeliveryTime(deliveryTime);
        param.setNotRefundStatus(1);
        return handleOrder(param, CONFIRM);
    }

    @Override
    public ServerResponseEntity handlerBalanceTimeStartOrder() {
        OrderSearchDTO param = new OrderSearchDTO();
        param.setStatus(OrderStatus.WAIT_BALANCE.value());
        param.setPreSaleType(PreSaleType.DEPOSIT.value());
        param.setMaxBalanceAmount(0L);
        param.setBalanceStartTime(new Date());
        return handleOrder(param, PRESALE);
    }

    @Override
    public ServerResponseEntity handleBalanceTimeExpireOrder() {
        OrderSearchDTO param = new OrderSearchDTO();
        param.setStatus(OrderStatus.WAIT_BALANCE.value());
        //定金预售
        param.setPreSaleType(PreSaleType.DEPOSIT.value());
        // 尾款金额大于0的订单
        param.setMinBalanceAmount(1L);
        param.setBalanceEndTime(new Date());
        return handleOrder(param, PRESALE_CANAL);
    }

    @Override
    public ServerResponseEntity checkPurchaseOrderPayStatus() {
        EsPageDTO esPageDTO = new EsPageDTO();
        esPageDTO.setPageSize(100);
        esPageDTO.setPageNum(1);
        PurchaseOrderSearchDTO purchaseOrderSearchDTO = new PurchaseOrderSearchDTO();
        // 采购订单状态（0:已作废   1:待入库   2:发货   3:部分入库   4:代付款   5:已完成   4:待审核）
        purchaseOrderSearchDTO.setStatus(4);
        purchaseOrderSearchDTO.setEndTime(new Date());
        EsPageVO<EsPurchaseOrderVO> esPurchaseOrderPage = purchaseOrderSearchManager.page(esPageDTO, purchaseOrderSearchDTO);
        List<Long> purchaseOrderIds = esPurchaseOrderPage.getList().stream().map(EsPurchaseOrderVO::getPurchaseOrderId).collect(Collectors.toList());
        while (esPurchaseOrderPage.getPages() > esPageDTO.getPageNum()) {
            esPageDTO.setPageNum(esPageDTO.getPageNum() + 1);
            esPurchaseOrderPage = purchaseOrderSearchManager.page(esPageDTO, purchaseOrderSearchDTO);
            purchaseOrderIds.addAll(esPurchaseOrderPage.getList().stream().map(EsPurchaseOrderVO::getPurchaseOrderId).collect(Collectors.toList()));
        }
        return purchaseOrderFeignClient.updatePurchaseOrderStatus(purchaseOrderIds, StatusEnum.DISABLE.value());
    }

    private ServerResponseEntity<Object> handleOrder(OrderSearchDTO param, String type) {
        param.setFetchSource(EsConstant.ORDER_TASK_FETCH_SOURCE);
        String scrollId = null;
        //设置查询超时时间
        Scroll scroll = new Scroll(TimeValue.timeValueMinutes(5L));
        SearchRequest searchRequest = orderSearchManager.buildScrollSearchRequest(param, scroll);
        try{
            // 进行第一次滚动查询
            SearchResponse searchResponse = EsSearchUtil.search(searchRequest);
            scrollId = searchResponse.getScrollId();
            /**
             *在这个位置已经读到了前一百条数据，可以在这先对这一百数据进行处理。下面滚动查询剩下的数据
             */
            sendOrderMq(searchResponse, type);
            //记录要滚动的ID
            //滚动查询部分，将从第1001笔数据开始取
            SearchHit[] hitsScroll = searchResponse.getHits().getHits();
            while (hitsScroll != null && hitsScroll.length > 0) {
                //构造滚动查询条件
                SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
                searchScrollRequest.scroll(scroll);
                searchResponse = EsSearchUtil.scroll(searchScrollRequest);
                scrollId = searchResponse.getScrollId();
                hitsScroll = searchResponse.getHits().getHits();
                // 发送消息
                sendOrderMq(searchResponse, type);
            }
        } catch (Exception e) {
            logger.error("订单结算异常: {}",e);
            return ServerResponseEntity.fail(ResponseEnum.EXCEPTION);
        } finally {
            if (Objects.nonNull(scrollId)) {
                //清除滚动，否则影响下次查询
                EsSearchUtil.clearScroll(scrollId);
            }
        }
        return ServerResponseEntity.success();
    }

    private void sendOrderMq(SearchResponse searchResponse, String type) {
        List<Long> orderIds = new ArrayList<>();
        for (SearchHit hit : searchResponse.getHits().getHits()) {
            EsOrderBO esOrderBO = JSON.parseObject(hit.getSourceAsString(), EsOrderBO.class);
            orderIds.add(esOrderBO.getOrderId());
        }
        if (CollUtil.isEmpty(orderIds)) {
            return;
        }

        // 发送mq消息
        if (StrUtil.equals(type, CONFIRM)) {
            ServerResponseEntity serverResponseEntity = orderFeignClient.confirmOrder(orderIds);
            if (serverResponseEntity.isFail()) {
                throw new LuckException(serverResponseEntity.getMsg());
            }
        } else if (StrUtil.equals(type, SETTLED)) {
            ServerResponseEntity serverResponseEntity = orderFeignClient.settledOrder(orderIds);
            if (serverResponseEntity.isFail()) {
                throw new LuckException(serverResponseEntity.getMsg());
            }
        } else if (StrUtil.equals(type, PRESALE)) {
            // 发送消息，订单支付成功
            SendStatus sendStatus = orderNotifyTemplate.syncSend(RocketMqConstant.ORDER_NOTIFY_TOPIC, new GenericMessage<>(new PayNotifyBO(orderIds, null,null))).getSendStatus();
            if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
                // 消息发不出去就抛异常，因为订单回调会有多次，几乎不可能每次都无法发送出去，发的出去无所谓因为接口是幂等的
                throw new LuckException(ResponseEnum.EXCEPTION);
            }
        } else if (StrUtil.equals(type, PRESALE_CANAL)) {
            // 发送消息，订单支付成功
            SendStatus sendStatus = presaleOrderCanalMqTemplate.syncSend(RocketMqConstant.PRESALE_ORDER_CANAL_TOPIC, new GenericMessage<>(orderIds)).getSendStatus();
            if (!Objects.equals(sendStatus, SendStatus.SEND_OK)) {
                // 消息发不出去就抛异常，因为订单回调会有多次，几乎不可能每次都无法发送出去，发的出去无所谓因为接口是幂等的
                throw new LuckException(ResponseEnum.EXCEPTION);
            }
        }
    }

}
